//package com.open.source.flink.cdc;
//
//import com.alibaba.fastjson.JSONObject;
//import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
//import io.debezium.data.Envelope;
//import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
//import org.apache.flink.api.common.typeinfo.TypeInformation;
//import org.apache.flink.util.Collector;
//import org.apache.kafka.connect.data.Field;
//import org.apache.kafka.connect.data.Struct;
//import org.apache.kafka.connect.source.SourceRecord;
//
///**
// * 自定义json反序列化类
// * @author ZonLen since on 2022/7/20 下午10:54
// */
//public class JsonDeserializationSchema implements DebeziumDeserializationSchema<JSONObject> {
//  private static final long serialVersionUID = 4186899475568749993L;
//
//
//  public JsonDeserializationSchema() {
//  }
//
//  @Override
//  public void deserialize(SourceRecord record, Collector<JSONObject> out) throws Exception {
//    Struct valueStruct = (Struct) record.value();
//    Struct sourceStrut = valueStruct.getStruct("source");
//    //获取数据库的名称
//    String database = sourceStrut.getString("db");
//    //获取表名
//    String table = sourceStrut.getString("table");
//
//    //获取类型（c-->insert,u-->update）
//    String type = Envelope.operationFor(record).toString().toLowerCase();
//    //read
//    //System.out.println("type = "+type);
//    if (type.equals("create")) {
//      type = "insert";
//    }
//
//
//    JSONObject json = new JSONObject();
//    json.put("database", database);
//    json.put("table", table);
//    json.put("type", type);
//
//    //获取数据data
//    Struct afterStruct = valueStruct.getStruct("after");
//    setData(json, "after", afterStruct);
//
//    Struct beforeStruct = valueStruct.getStruct("before");
//    setData(json, "before", beforeStruct);
//
//    //向下游传递数据
//    out.collect(json);
//  }
//
//  private void setData(JSONObject json, String type, Struct struct) {
//    JSONObject dataJson = new JSONObject();
//    if (struct != null) {
//      for (Field field : struct.schema().fields()) {
//        String fieldName = field.name();
//        Object fieldValue = struct.get(field);
//        dataJson.put(fieldName, fieldValue);
//      }
//    }
//    json.put(type, dataJson);
//  }
//
//  @Override
//  public TypeInformation<JSONObject> getProducedType() {
//    return BasicTypeInfo.of(JSONObject.class);
//  }
//}